package com.pinoc.demo.controller;

import com.alibaba.fastjson.JSON;
import com.pinoc.demo.common.es.EsRestHighLevelClient;
import com.pinoc.demo.domain.es.EsUserDocument;
import com.pinoc.demo.service.es.EsOperateService;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.ResponseException;
import org.elasticsearch.client.ml.job.config.RuleCondition.AppliesTo;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import com.pinoc.demo.manager.CatManager;
import com.pinoc.demo.manager.StudentManager;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author yinpeng10
 */
@RestController
@RequestMapping("")
public class HelloWorldController {
	public static final String index_name = "demo_index_3";

	@RequestMapping("/welcome")
	public String index() {
		return "welcome...";
	}

	@RequestMapping("/hello")
	public String hello() {
		return "hello...";
	}

	@Autowired
	private EsRestHighLevelClient client;
	@Autowired
	private EsOperateService esOperateService;

	@RequestMapping("es/test")
	public void esTest() throws InterruptedException {
		ThreadPoolExecutor executor = new ThreadPoolExecutor(8, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue(256));
		int count = 130;
		final AtomicInteger atomicInteger = new AtomicInteger(0);
		for (int i = 1; i <= count; i++) {
			executor.execute(() -> {
				UpdateResponse updateResponse = null;
				try {
					//更新and+1
					 doUpdate(atomicInteger, null,null, null);
				} catch (ElasticsearchStatusException e) {
					int retryCount = 1;
					while(retryCount < 10) {
						try {
							GetResponse byId = esOperateService.getById(index_name, 1 + "");
							long seqNo = byId.getSeqNo();
							long primaryTerm = byId.getPrimaryTerm();
							updateResponse = doUpdate(atomicInteger, atomicInteger.get(), seqNo, primaryTerm);
							if(updateResponse != null) {
								if(updateResponse.getShardInfo().getSuccessful() == 1) {
									break;
								}
							}
						} catch (Exception ex) {
							retryCount++;
						}
					}
				} catch (Exception e) {
					System.out.println("error....");
				}
			});
		}
		executor.shutdown();
		Thread.sleep(3000);
		System.out.println("result: " + atomicInteger);
	}

	private UpdateResponse doUpdate(AtomicInteger atomicInteger, Integer age, Long seqNo, Long primaryTerm) throws IOException {
		UpdateRequest request = new UpdateRequest(index_name, 1 + "");
		EsUserDocument user = new EsUserDocument();
		user.setId(1L);
		if(age != null) {
			user.setAge(atomicInteger.get());
		} else {
			user.setAge(atomicInteger.incrementAndGet());
		}
		request.doc(JSON.toJSONString(user), XContentType.JSON);
		if(seqNo != null) {
			request.setIfSeqNo(seqNo);
		}
		if(primaryTerm != null) {
			request.setIfPrimaryTerm(primaryTerm);
		}
		return client.getClient().update(request, RequestOptions.DEFAULT);
	}
	@RequestMapping("es/init")
	public void esTest2() throws IOException {
		UpdateRequest request = new UpdateRequest(index_name, 1 + "");
		EsUserDocument user = new EsUserDocument();
		user.setId(1L);
		user.setAge(0);
		request.doc(JSON.toJSONString(user), XContentType.JSON).upsert(JSON.toJSONString(user), XContentType.JSON);
		client.getClient().update(request, RequestOptions.DEFAULT);
	}

	@RequestMapping("es/testAdd")
	@ResponseBody
	public String testAdd() throws IOException, InterruptedException {
		ThreadPoolExecutor executor = new ThreadPoolExecutor(8, 16, 60, TimeUnit.SECONDS, new LinkedBlockingQueue(256));
		int threadCount = 200;
		final AtomicInteger atomicInteger = new AtomicInteger(0);
		for (int i = 1; i <= threadCount; i++) {
			executor.execute(() -> {
				for (int i1 = 0; i1 < 200; i1++) {
					try {
						GetResponse byId = esOperateService.getById(index_name, 1 + "");
						long seqNo = byId.getSeqNo();
						long primaryTerm = byId.getPrimaryTerm();
						doUpdate(atomicInteger, atomicInteger.get(), seqNo, primaryTerm);
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			});
		}
		executor.shutdown();
		Thread.sleep(3000);
		System.out.println("num: "+ atomicInteger);
		return atomicInteger.toString();
	}
}
